/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package br.ufu.sdg12014.thriftpluresdb.server;

import br.ufu.sdg12014.base.clientreply.InvalidOperation;
import br.ufu.sdg12014.base.clientreply.ThriftServer;
import br.ufu.sdg12014.base.replycoordinator.ThriftReplica;
import br.ufu.sdg12014.base.replycoordinator.Transacao;
import br.ufu.sdg12014.thriftpluresdb.client.DBNode;
import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.thrift.TException;
import persistence.FilePersistence;
import plures.Core;
import plures.Return;
import plures.transaction.Command;
import plures.transaction.Commands;
import plures.transaction.Transaction;
import util.Pair;

/**
 *
 * @author SergioPeres
 */
public class ServerHandler extends FilePersistence implements ThriftServer.Iface, ThriftReplica.Iface
{
    private List<DBNode> participants;
    private Core core;
    private HashSet<Integer> tokens;
    private transient HashMap<Integer, Transaction> transactions;
    private boolean lock;
    private final static int LONG_TIMEOUT = 3000;
    private final static int SHORT_TIMEOUT = 1000;
    private ArrayList<Pair<Integer,Transaction>> committed;
    private int counter;

    public ServerHandler(String dbFile, int dataLength) throws FileNotFoundException, IOException, TException
    {
        super(dbFile);
        this.tokens = new HashSet<>();
        this.participants = new ArrayList<>();
        this.transactions = new HashMap<>();
        this.committed = new ArrayList<>();
        this.lock = false;
        
        this.core = new Core(dataLength);
        this.counter = 0; 
    }
    
    public static ServerHandler restore(String restoreFile) throws UnknownHostException
    {
        ServerHandler sh = (ServerHandler) FilePersistence.load(restoreFile);
        sh.core.initValueLength();
        sh.transactions = new HashMap<>();
        for(DBNode n : sh.participants)
        {
            n.init();
        }
        sh.notifyNodes();
        DBNode n = sh.participants.get(0);
        List<Transacao> transacoes = n.getCommitted(sh.counter);
        for(Transacao transacao :  transacoes)
        {
            System.out.print("\tRetrieving: transaction #" + (sh.counter + 1) + "... ");
            sh.commit(transacao.clientToken, new Transaction(transacao.clientToken, transacao));
            System.out.println("Done");
        }
        
        return sh;
    }
    
    public void init(String nodesFileName) throws FileNotFoundException, IOException
    {
        BufferedReader bf = new BufferedReader(new FileReader(nodesFileName));
        for(String line = bf.readLine(); line != null; line = bf.readLine())
        {
            DBNode r = new DBNode(line);
            this.participants.add(r);
        }
        this.notifyNodes();
    }
    
    public void notifyNodes() throws UnknownHostException
    {
        ArrayList<String> up = new ArrayList<>();
        up.add("add");
        up.add(InetAddress.getLocalHost().getHostAddress());
        for(DBNode reply :  this.participants)
        {
            reply.updateServerFile(up);
        }
    }

    //<editor-fold defaultstate="collapsed" desc="Create, Read, Update, Delete">
    @Override
    public String opCreate(int id, int chave, String valor) throws InvalidOperation, TException
    {
        if (!this.checkAuthorizationInTransacao(id))
        {
            return unAuthError;
        }
        Return r = core.checkCreate(chave, valor, id);
//        if (r.getStatus() == 200)
//        {
        ArrayList<Object> args = new ArrayList<>();
        args.add(chave);
        args.add(valor);
        Command c = new Command(Commands.CREATE, args);
        this.transactions.get(id).add(c);
//        }
        return r.toString();
    }

    @Override
    public String opRead(int id, int chave) throws InvalidOperation, TException
    {
        if (!this.checkAuthorizationInTransacao(id))
        {
            return unAuthError;
        }
        Return r = core.checkRead(chave, id);
//        if (r.getStatus() == 200)
//        {
        ArrayList<Object> args = new ArrayList<>();
        args.add(chave);
        Command c = new Command(Commands.READ, args);
        this.transactions.get(id).add(c);
//        }
        return chave + " -> " + r.toString();
    }

    @Override
    public String opUpdate(int id, int chave, String valor) throws InvalidOperation, TException
    {
        if (!this.checkAuthorizationInTransacao(id))
        {
            return unAuthError;
        }
        Return r = core.checkUpdate(chave, valor, id);
//        if (r.getStatus() == 200)
//        {
        ArrayList<Object> args = new ArrayList<>();
        args.add(chave);
        args.add(valor);
        Command c = new Command(Commands.UPDATE, args);
        this.transactions.get(id).add(c);
//        }
        return chave + " -> " + r.toString();
    }

    @Override
    public String opDelete(int id, int chave) throws InvalidOperation, TException
    {
        if (!this.checkAuthorizationInTransacao(id))
        {
            return unAuthError;
        }
        Return r = core.checkDelete(chave, id);
//        if (r.getStatus() == 200)
//        {
        ArrayList<Object> args = new ArrayList<>();
        args.add(chave);
        Command c = new Command(Commands.DELETE, args);
        this.transactions.get(id).add(c);
//        }
        return r.toString();
    }
//</editor-fold>

    public boolean opBegin(int client) throws InvalidOperation, TException
    {
        if (!this.checkAuthorization(client))
        {
            return false;
        }
        Transaction transaction = new Transaction(client);
        this.transactions.put(client, transaction);
        return true;
    }

    public synchronized String opEnd(int client) throws InvalidOperation, TException
    {
        if (!this.checkAuthorizationInTransacao(client))
        {
            return unAuthError;
        }
        Transaction transaction = this.transactions.get(client);
        Transacao transacao = transaction.toTransacao();
        try
        {
            boolean flag = this.sendVoteRequest(client, transacao);

            // se alguém não concordou com o commit, então aborta-se a operação
            if (!flag)
            {
                this.execute(client, DBNode.Operation.GLOBAL_ABORT);
                return abortError;
            }

            this.execute(client, DBNode.Operation.PREPARE_COMMIT);
            this.execute(client, DBNode.Operation.GLOBAL_COMMIT);

        } catch (InterruptedException ex)
        {
            ex.printStackTrace();
            return error + " - InterruptedException";
        } finally
        {
            ArrayList<DBNode> removeList = new ArrayList<>(); // remover as réplicas que 'caíram'
            for (DBNode r : this.participants)
            {
                if (!r.connect)
                {
                    removeList.add(r);
                }
            }
            ArrayList<String> list = new ArrayList<>();
            list.add("rem");
            for(int i = 0; i < removeList.size(); i++)
            {
                DBNode disconnected = removeList.get(i);
                list.add(disconnected.address);
                this.opUpdateServerFile(list);
                list.remove(1);
            }
            for(DBNode connected : this.participants)
            {
                for(DBNode disconnected : removeList)
                {
                    list.add(disconnected.address);
                    connected.updateServerFile(list);
                    list.remove(1);
                }
            }
        }
        return commitOK;
    }

    private boolean sendVoteRequest(int client, Transacao transacao) throws InterruptedException, TException
    {
        boolean flag = true;

        if (!this.opRequestVote(client, transacao)) // enviar pedido de voto para ele mesmo
        {
            System.out.println("This -> [No]");
            return false;
        }
        for (DBNode reply : this.participants)
        {
            if(!reply.requestVote(client, transacao))
            {
                System.out.println(reply.address + " -> [No]");
                flag = false;
                break;
            }
            System.out.println(reply.address + " -> [Yes]");
        }

//        // verificar repostas das outras réplicas
//        for (DBNode r : this.participants)
//        {
//            if (r.connect && (r.getOperationResponse() == -1))
//            {
//                flag = false;
//                System.out.println(r.address + " -> [No]");
//                break;
//            } else if (r.connect && (r.getOperationResponse() == 0))
//            {
//                r.connect = false;
//            }
//            System.out.println(r.address + " -> [Yes]");
//        }
        return flag;
    }

    private void execute(int client, DBNode.Operation operation) throws InterruptedException, TException
    {
        ExecutorService es = Executors.newCachedThreadPool();
        for (DBNode reply : this.participants)
        {
            reply.setOperation(operation, client);
            es.execute(reply);
        }
        switch (operation)
        {
            case PREPARE_COMMIT:
                this.opPrepareCommit(client);
                break;
            case GLOBAL_ABORT:
                this.opGlobalAbort(client);
                break;
            case GLOBAL_COMMIT:
                this.opGlobalCommit(client);
                break;
        }
        this.wait(SHORT_TIMEOUT);
        es.shutdownNow();
    }

    private boolean checkAuthorization(int client) throws InvalidOperation, TException
    {
        return this.tokens.contains(client) && !this.transactions.containsKey(client);
    }

    private boolean checkAuthorizationInTransacao(int client) throws InvalidOperation, TException
    {
        return this.tokens.contains(client) && this.transactions.containsKey(client);
    }

    public boolean opRequestVote(int client, Transacao transacao) throws InvalidOperation, TException
    {
        Transaction transaction = new Transaction(client, transacao);
        if (this.lock)
        {
            return false;
        }
        Commands sig;
        boolean permission = false;
        for (Command cmd : transaction.getCmds())
        {
            sig = cmd.getSignature();

            switch (sig)
            {
                case CREATE:
                    permission = this.core.canCreate(cmd.getKey(), cmd.getValue(), client);
                    break;
                case READ:
                    permission = this.core.canRead(cmd.getKey(), client);
                    break;
                case UPDATE:
                    permission = this.core.canUpdate(cmd.getKey(), client);
                    break;
                case DELETE:
                    permission = this.core.canDelete(cmd.getKey(), client);
                    break;
            }
            if (!permission)
            {
                return false;
            }
        }
        this.transactions.put(client, transaction);
        this.lock = true;
        return true;
    }

    public boolean opPrepareCommit(int client) throws InvalidOperation, TException
    {
        return true;
    }

    public boolean opGlobalCommit(int client) throws InvalidOperation, TException
    {
        return this.commit(client, this.transactions.get(client));
    }
    
    public boolean commit(int client, Transaction transaction)
    {
        Commands sig;
        String str = "";
        System.out.println("\nTransaction: client #" + client);
        for (Command cmd : transaction.getCmds())
        {
            sig = cmd.getSignature();

            switch (sig)
            {
                case CREATE:
                    this.core.create(cmd.getKey(), cmd.getValue(), client);
                    System.out.println("\t create: "+ cmd.getKey() + " -> " + cmd.getValue());
                    break;
                case READ:
                    str = this.core.read(cmd.getKey(), client) + "";
                    System.out.println("\t read: "+ cmd.getKey() + " = " + str);
                    break;
                case UPDATE:
                    str = this.core.update(cmd.getKey(), cmd.getValue(), client) + "";                    
                    System.out.println("\t update: "+ cmd.getKey() + " -> " + cmd.getValue() + " = " + str);
                    break;
                case DELETE:
                    this.core.delete(cmd.getKey(), client);                    
                    System.out.println("\t delete: "+ cmd.getKey());
                    break;
            }
        }
        System.out.println("Committed");
        this.transactions.remove(client);
        this.lock = false;
        FilePersistence.save(this);
        this.committed.add(new Pair(client,transaction));
        this.counter++;
        return true;
    }

    public boolean opGlobalAbort(int client) throws InvalidOperation, TException
    {
        this.lock = false;
        this.transactions.remove(client);
        return true;
    }

    public String opEnter() throws InvalidOperation, TException
    {
        return "OK";
    }

    public String opUpdateServerFile(List<String> list) throws InvalidOperation, TException
    {
        if(list.get(0).equals("add"))
        {
            this.addServer(new DBNode(list.get(1)));
            System.out.println("[NET] This --> : Node (" + list.get(1) + ") [ OK ]");
        }
        else if(list.get(0).equals("rem"))
        {
            this.remServer(list.get(1));
            System.out.println("[NET] Node (" + list.get(1) + ") removed [ OK ]");
        }

        return "OK";
    }
    
    public void remServer(String address) throws InvalidOperation, TException
    {
        for(Iterator<DBNode> it = this.participants.iterator(); it.hasNext();)
        {
            DBNode n = it.next();
            if(n.address.equals(address))
            {
                n.disconnect();
                it.remove();
                n = null;
            }
        }
        System.gc();
    }
    
    public List<Transacao> getCommitted(int counter)
    {
        ArrayList<Transacao> out = new ArrayList<>();
        for(int i = counter + 1; i < this.committed.size(); i++)
        {
            Pair<Integer,Transaction> p = this.committed.get(i);
            Transacao t = p.second.toTransacao();
            out.add(t);
        }
        return out;
    }
    
    
    public boolean updateData(List<Transacao> transacoes) throws TException
    {
        int client = Integer.parseInt(transacoes.get(0).ops.get(0).argumentos.get(0));
        for(Transacao t :  transacoes)
        {
            this.commit(client, new Transaction(client,t));
        }
        return true;
    }

    private void addServer(DBNode c) throws InvalidOperation, TException
    {
        this.participants.add(c);
    }

    @Override
    public int opGetToken() throws InvalidOperation, TException
    {
        int token;
        do
        {
            token = (int) (10000000.0 * Math.random());
        } while (this.tokens.contains(token));
        this.tokens.add(token);
        return token;
    }

    private static final String unAuthError = "[ERROR] Unauthorized";
    private static final String abortError = "[INFO] COMMIT: Global Abort";
    private static final String commitOK = "[INFO] COMMIT: Committed";
    private static final String error = "[ERROR] COMMIT: Synchronization error";
}
